package com.mtiiot.mq.service;

import com.mtiiot.mq.esdao.BaseElasticsearch;
import com.mtiiot.mq.utils.Contants;

import org.codehaus.jackson.map.ObjectMapper;

import java.util.Map;

/**
 * Created by xyr on 2016/11/1.
 * es 服务类
 */
public class ElasticsearchService {

    private LogService1 logService1;
    private BaseElasticsearch baseElasticsearch;

    // 处理kafka中的es队列
    public  void handleEsData(String key,String value){
        // time$$_$$index$$_$$type$$_$$insert$$_$$id
        String[] arr = key.split("$$_$$");
        if (arr.length >= 4){
            String index = arr[1].trim();
            String type = arr[2].trim();
            String handle = arr[3].trim();
            if (handle.equals("insert")){
                this.createIndexByAsy(index,type,value);
            } else if (handle.equals("update")){
                String id = arr[4];
                updateIndexByAsy(index,type,id,value);
            } else if (handle.equals("delete")){
                deleteIndexByAsy(index,type,value);
            }
        }
    }

    /**
     * 创建一个文档
     */
    public void createIndexByAsy(String index,String type,String sourceContent){
        try {
            baseElasticsearch.createIndex(index,type,sourceContent);
        } catch (Exception e){
            logService1.sendLog(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "ElasticsearchService", "createIndexByAsy"), LogLevel.ERROR, "createIndexByAsy error", e.getMessage());
        }
    }

    /**
     * 更新一个文档
     */
    public void updateIndexByAsy(String index,String type,String id,String sourceContent){
        try {
            //将json 转成map
            ObjectMapper mapper = new ObjectMapper();
            Map<String,String> map = mapper.readValue(sourceContent,Map.class);
            baseElasticsearch.updateIndex(index,type,id,map);
        } catch (Exception e){
            logService1.sendLog(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "ElasticsearchService", "updateIndexByAsy"), LogLevel.ERROR, "updateIndexByAsy error", e.getMessage());
        }
    }

    /**
     * 删除一个文档
     */
    public void deleteIndexByAsy(String index, String type, String id){
        try {
            baseElasticsearch.deleteIndex(index, type, id);
        } catch (Exception e){
            logService1.sendLog(new MetadataInfo(Contants.SYS_ID_FS, Contants.FSTASK_HOST, Contants.MODULE_FSTASK, "ElasticsearchService", "deleteIndexByAsy"), LogLevel.ERROR, "deleteIndexByAsy error", e.getMessage());
        }
    }




    public LogService1 getLogService1() {
        return logService1;
    }

    public void setLogService1(LogService1 logService1) {
        this.logService1 = logService1;
    }

    public BaseElasticsearch getBaseElasticsearch() {
        return baseElasticsearch;
    }

    public void setBaseElasticsearch(BaseElasticsearch baseElasticsearch) {
        this.baseElasticsearch = baseElasticsearch;
    }
}
